package com.corn.tutorial.reactive.lession2;

import com.corn.tutorial.reactive.base.Book;
import com.corn.tutorial.reactive.base.InMemoryDataSource;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2020/12/15 10:34
 */

public class ReactorCoreApiTest {

    public static void main(String[] args) {
        // 测试生成flux
//        createFlux();
        // subscribe方法测试
//        testSubscribe();
        // 异步创建Mono
//        createMonoAsync();
        // map and flatMap
//        testMapAndFlatMap();
        // 流程API
//        testThen();
        // mono 和 flux 的转换
//        testMonoFluxExchange();
        // 异常处理
//        testErrorHandler();
        // flux Collect API
//        testFluxCollectApi();
        // flux create api
        createFluxAsync();
    }

    /**
     * 创建flux
     */
    private static final void createFlux() {
        Flux<Integer> fluxByJust = Flux.just(1, 2, 3, 4, 5);
        printFlux("just", fluxByJust);

        Flux<Integer> fluxByArray = Flux.fromArray(new Integer[]{1, 2, 3, 4, 5});
        printFlux("fromArray", fluxByArray);

        Flux<Integer> fluxByIterable = Flux.fromIterable(Arrays.asList(1, 2, 3, 4, 5));
        printFlux("fromIterable", fluxByIterable);

        Flux<Integer> fluxByStream = Flux.fromStream(Stream.of(1, 2, 3, 4, 5));
        printFlux("fromStream", fluxByStream);

        Flux<Integer> fluxByRange = Flux.range(3, 5);
        printFlux("range", fluxByRange);

        AtomicInteger i = new AtomicInteger(1);
        Flux<Integer> fluxByGenerate = Flux.generate(sink -> {
            if (i.get() <= 5) {
                sink.next(i.getAndIncrement());
            } else {
                sink.complete();
            }
        });
        printFlux("fluxByGenerateWithOutInitValue", fluxByGenerate);

        /**
         * 参数1: callable 返回一个generator的初始值
         * 参数2: BiFunction<S, SynchronousSink<T>, S> S跟callable的类型一致 表示初始值类型和返回值类型
         * SynchronousSink<T> 用于放flux中装数据的对象 sink.next()装 sink.complete()完成
         */
        Flux<Object> fluxGenerateWithInitValue = Flux.generate(() -> 1, (preState, sink) -> {
            if (preState <= 5) {
                sink.next(preState);
                return preState + 1;
            } else {
                sink.complete();
            }
            return 0;
        });
        printFlux("fluxGenerateWithInitValue", fluxGenerateWithInitValue);

    }

    /**
     * subscribe 各种方法api
     */
    private static final void testSubscribe() {
        Flux<Integer> flux = Flux.range(1, 5);
        // do nothing 什么都不做
        flux.subscribe();
        // Do something with each produced value.  subscribe中消费每次生成的最后结果
        flux.subscribe(t -> System.out.print(t + "->"));
        System.out.println("================");
        //	Deal with values but also react to an error. 带异常捕获处理
        flux.subscribe(t -> {
            if (t > 4) {
                throw new RuntimeException(" num is greater than 4");
            }
            System.out.print(t + "->");
        }, error -> {
            error.printStackTrace();
        });
        System.out.println("================");

        // Deal with values and errors but also run some code when the sequence successfully completes.
        // 带异常处理 且 最后可以有一个runnable 做收尾处理
        flux.subscribe(t -> {
            System.out.print(t + "->");
        }, e -> {
            System.out.println(e.getCause());
        }, () -> {
            System.out.println("done");
        });

        // Deal with values and errors and successful completion but also do something with the Subscription produced by this subscribe call.
        // 带异常处理 + 收尾处理 + 被压请求
        flux.subscribe(t -> {
            System.out.print(t + "->");
        }, e -> {
            System.out.println(e.getCause());
        }, () -> {
            System.out.println("done");
        }, subscription -> subscription.request(3));
    }

    /**
     * 异步创建 Mono
     */
    private static void createMonoAsync() {
        Mono<String> monoFromCallable = Mono.fromCallable(() -> Thread.currentThread().getName() + " " + LocalDate.now()).publishOn(Schedulers.elastic());
        blockMono(monoFromCallable);
        Mono<Object> monoFromRunnable = Mono.fromRunnable(() -> System.out.println(Thread.currentThread().getName() + " " + LocalDate.now())).publishOn(Schedulers.elastic());
        blockMono(monoFromRunnable);
        Mono<String> monoBySupplier = Mono.fromSupplier(() -> Thread.currentThread().getName() + " " + LocalDate.now()).publishOn(Schedulers.elastic());
        blockMono(monoBySupplier);
    }

    /**
     * create api 异步多线程生成
     */
    private static void createFluxAsync() {
        Flux<Integer> createFluxByCreate = Flux.create(sink -> {
            // 通常调用外部api 生成数据调用next
            IntStream.rangeClosed(1, 5).forEach(sink::next);
            sink.complete();
            ;
        });
        printFlux("createFluxByCreate", createFluxByCreate);
    }

    private static void blockMono(Mono mono) {
        mono.doOnNext(System.out::println).block();
    }

    /**
     * map 和 flatMap 区别主要在于 map 返回真实类型 flatMap需要返回Publisher 如Mono
     */
    private static void testMapAndFlatMap() {
        Flux.range(1, 5).map(i -> {
            System.out.print("map " + i);
            return i;
        }).doOnComplete(System.out::println).subscribe();

        Flux.range(1, 5).flatMap(i -> {
            System.out.print("flatMap " + i);
            return Mono.just(i);
        }).doOnComplete(System.out::println).subscribe();
    }

    /**
     * thenReturn 返回一个publiser
     * thenMany 返回一个flux或者mono
     * thenEmpty 返回一个空的 Mono 或者 Flux
     */
    private static void testThen() {
        Mono<String> mono = Mono.just("hello")
                .doOnNext(s -> {
                    s = "world " + s;
                    System.out.println(s);
                })
                .doOnTerminate(System.out::println)
                .thenReturn("hahah");
        blockMono(mono);
        Flux<Integer> integerFlux = Mono.just(1)
                .doOnTerminate(() -> System.out.println("6666"))
                .thenMany(Flux.range(1, 5));
        printFlux("thenMany ", integerFlux);

        Mono<Void> voidMono = Flux.range(1, 3)
                .doOnNext(System.out::print)
                .doOnComplete(System.out::println)
                .thenEmpty(Mono.empty());
        blockMono(voidMono);
    }

    /**
     * Mono to Flux Mono#flux()
     * Flux to Mono<List<T>> Flux转Mono Flux#collectList
     */
    private static void testMonoFluxExchange() {
        Flux<Integer> monoToFlux = Mono.just(1).flux();
        printFlux("monoToFlux: ", monoToFlux);
        Mono<List<Integer>> fluxToMono = Flux.range(1, 3).collectList();
//        fluxToMono.doOnNext(System.out::println).block();
        fluxToMono.doOnNext(list -> {
            list.forEach(System.out::println);
        }).block();
    }

    private static void testErrorHandler() {
        // 方式1: 直接在 publisher 内部抛出异常
//        Flux.range(1, 5)
//                .map(i -> {
//                    if (i > 3) {
//                        throw new RuntimeException("greater than 3");
//                    }
//                    return i;
//                }).subscribe();

        // 方式2: 使用 publiser中的api
        Flux.range(1, 5)
                .flatMap(i -> {
                    if (i > 3) {
                        return Mono.error(new RuntimeException("greater then 3"));
                    }
                    return Mono.just(i);
                }).subscribe();
    }

    /**
     * 测试 Flux中的 collectAPI
     */
    private static void testFluxCollectApi() {
        // collectSortedList 排序的list
//        Flux.fromArray(InMemoryDataSource.books)
//                .collectSortedList(Comparator.comparing(Book::getPrice).reversed())
//                .doOnNext(list -> {
//                    list.forEach(System.out::println);
//                }).block();

//        Flux.fromArray(InMemoryDataSource.books)
//                .sort(Comparator.comparing(Book::getPrice).reversed())
//                .doOnNext(System.out::println)
//                .subscribe();
//        Flux.fromArray(InMemoryDataSource.books)
//                .collectMultimap(Book::getCategory)
//                .map(groupedMap -> {
//                    Flux<Book> bookFlux = Flux.fromIterable(groupedMap.values())
//                            .flatMap(books -> {
//                                        return Flux.fromIterable(books).sort(Comparator.comparing(Book::getPrice).reversed()).next();
//                                    }
//                            );
//                    return bookFlux;
//                })
//                .doOnNext(t->{
//                    printFlux("sorted book ",t);
//                })
//                .subscribe();


//        Flux.fromArray(InMemoryDataSource.books)
//                .collectMap(Book::getCategory)
//                .doOnNext(System.out::println)
//                .subscribe();

        Flux.fromArray(InMemoryDataSource.books)
                .collectMultimap(Book::getCategory)
                .map(groupMap -> {
                    Map<String, Book> sortedMap = new HashMap<>();
                    groupMap.entrySet().forEach(kv -> {
                        Book max = kv.getValue().stream().max(Comparator.comparing(Book::getPrice)).get();
                        sortedMap.put(kv.getKey(), max);
                    });
                    return sortedMap;
                })
                .doOnNext(System.out::println)
                .subscribe();
    }

    /**
     * 打印流
     *
     * @param name
     * @param flux
     */
    private static void printFlux(String name, Flux<?> flux) {
        flux.doOnSubscribe(t -> System.out.print(name + ": ")).
                doOnNext(i -> System.out.print(i + "->"))
                .doOnError(Throwable::printStackTrace)
                .doOnComplete(System.out::println).subscribe();
    }
}
